package com.gis507.mapreduce.betterinter;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.Utils.UtilsClass;
import com.google.common.collect.EvictingQueue;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.ObjectUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class AISInterpolationReducer extends Reducer<Text, AisClassifyBean,AisClassifyBean, NullWritable> {

//    //设定的最大时间
//    private static final double STANDTIME = 30;
//    private static final double MinTime = 30;
//    //设定的最大距离
//    private static final double STANDSTANCE = 1000;
//    private static int sum = 0;
    private double maxStance = 500;

    @Override
    protected void reduce(Text key, Iterable<AisClassifyBean> values, Context context) throws IOException, InterruptedException {



            //实例化一个列表，用于存放数据
            List<AisClassifyBean> aisList = new ArrayList<>(10000);

            //遍历values
            for (AisClassifyBean value : values) {

                //将value添加到列表
                aisList.add(UtilsClass.getNewAisClassifyBean(value));

            }

            log.info("aisList的长度是：" + aisList.size());

        ArrayList<AisClassifyBean> allAisInterList = new ArrayList<>();
        ArrayList<AisClassifyBean> rightAisList = new ArrayList<>();
        for (int i = 0; i < aisList.size() - 2; i++){
                if (UtilsClass.getStance(aisList.get(i),aisList.get(i+1)) < 10000){
                    if (UtilsClass.getStance(aisList.get(i),aisList.get(i+1)) < maxStance){

                        for (int j = i + 1;j < aisList.size() - 2;j++){

                            if (UtilsClass.getStance(aisList.get(i),aisList.get(j)) > maxStance){

                                i= j - 1;
                                log.info("去除了密集的点，成功选取出两个有用点！");
                                rightAisList.add(UtilsClass.getNewAisClassifyBean(aisList.get(j)));
                                //直接写出
//                                context.write(aisList.get(j), NullWritable.get());
                                break;
                            }
                        }
                    }else{

                        ArrayList<AisClassifyBean> aisInterList = UtilsClass.insertPoint(aisList.get(i), aisList.get(i + 1),maxStance);
                        //将本次插值的数据放到总的插值列表中
                        allAisInterList.addAll(aisInterList);
                        //直接写出
//                        for (AisClassifyBean aisClassifyBean : aisInterList) {
//                            context.write(aisClassifyBean,NullWritable.get());
//                        }
                    }
                }else{

                    log.info("这是一个异常点");
                }


            }

            //汇总数据
            rightAisList.addAll(allAisInterList);
//
//            //写出
            for (AisClassifyBean aisInterBean : rightAisList) {

                context.write(aisInterBean,NullWritable.get());
            }


    }
}
